package job

import (
	"context"
	"github.com/ecodeclub/ekit/syncx/atomicx"
	rlock "github.com/gotomicro/redis-lock"
	"math/rand"
	"sync"
	"time"
	"webook/internal/service"
	"webook/pkg/logger"
)

type RankingJob struct {
	svc     service.RankingService
	l       logger.LoggerV1
	timeout time.Duration
	client  *rlock.Client
	key     string

	// 用sync.Mutex保护分布式锁，因为抢锁和续约可能导致并发问题
	localLock *sync.Mutex
	// 为了扩大锁的范围
	lock *rlock.Lock

	//job9
	load         *atomicx.Value[int32] //当前节点的负载情况
	podName      string                //当前节点的name 唯一标识
	maxAllowLoad int32                 //最大允许的负载 小于该值时才去运行任务
}

func NewRankingJob(svc service.RankingService, l logger.LoggerV1, client *rlock.Client,
	timeout time.Duration, load int32, podName string) *RankingJob {
	return &RankingJob{svc: svc,
		key:          "job:ranking",
		l:            l,
		client:       client,
		localLock:    &sync.Mutex{},
		timeout:      timeout,
		load:         atomicx.NewValueOf[int32](load),
		podName:      podName,
		maxAllowLoad: 80,
	}
}

func (r *RankingJob) SetLoad(load int32) {
	r.load.Store(load)
}
func (r *RankingJob) GetLoad() int32 {
	return r.load.Load()
}
func (r *RankingJob) TimedRefreshLoad() {
	ticker := time.NewTicker(time.Minute)
	defer ticker.Stop()
	// 使用for循环监听Ticker的C通道
	for _ = range ticker.C {
		nowLoad := int32(rand.Intn(100))
		r.SetLoad(nowLoad)
		ctx, cancel := context.WithTimeout(context.Background(), time.Millisecond*500)
		err := r.svc.UpdateLoadStatus(ctx, r.podName, nowLoad)
		if err != nil {
			r.l.Error("刷新负载情况失败", logger.String("节点名称", r.podName),
				logger.Int32("当前负载", nowLoad), logger.Error(err))
		}
		cancel() //循环里不要用defer
	}
}

func (r *RankingJob) Name() string {
	return "ranking"
}

func (r *RankingJob) Run() error {
	r.localLock.Lock()
	lock := r.lock
	if lock == nil {
		//job9 == start
		//如果该节点的负载高于最大可允许的负载 放弃抢占分布式锁
		if r.GetLoad() >= r.maxAllowLoad {
			return nil
		}
		//获取当前负载最小的节点和负载值
		ctx1, cancel1 := context.WithTimeout(context.Background(), time.Second)
		defer cancel1()
		minPodNames, minLoad, err := r.svc.GetMinLoadPod(ctx1)
		if err != nil { //获取节点值失败
			r.l.Error("获取负载最小的节点状况失败", logger.Error(err))
			return nil
		}
		minPodNameMap := make(map[string]struct{}, len(minPodNames))
		for _, elem := range minPodNames {
			minPodNameMap[elem] = struct{}{}
		}
		//不是负载最小的两个节点 且 当前负载大于等于两倍的最小负载 就返回
		if _, ok := minPodNameMap[r.podName]; !ok && r.GetLoad() >= 2*minLoad {
			return nil
		} else if ok && r.GetLoad() >= 2*minLoad { //是负载最小的节点但是最新的负载大于等于两倍的最小负载，也返回
			return nil
		}
		//抢到分布式锁的条件是：是负载最小的两个节点并且当前负载<最小负载的2倍
		//job9 == end
		// 抢分布式锁
		ctx, cancel := context.WithTimeout(context.Background(), time.Second*4)
		defer cancel()
		lock, err := r.client.Lock(ctx, r.key, r.timeout,
			&rlock.FixIntervalRetry{
				Interval: time.Millisecond * 100,
				Max:      3,
				// 重试的超时
			}, time.Second)
		if err != nil {
			//说明没有拿到分布式锁 说明大概率别的节点拿到了分布式锁 后续这个节点也会计算榜单
			r.localLock.Unlock()
			r.l.Warn("获取分布式锁失败", logger.Error(err))
			return nil
		}
		r.lock = lock
		r.localLock.Unlock()
		//自动续约 延长分布式锁的过期时间
		go func() {
			// 并不是非得一半就续约
			er := lock.AutoRefresh(r.timeout/2, r.timeout)
			if er != nil {
				// 续约失败了
				// 你也没办法中断当下正在调度的热榜计算（如果有）
				r.localLock.Lock()
				r.lock = nil
				//lock.Unlock()
				r.localLock.Unlock()
			}
		}()
	}
	// 这边就是你拿到了锁
	ctx, cancel := context.WithTimeout(context.Background(), r.timeout)
	defer cancel()
	//job9
	//拿到锁后发现负载过高 解锁
	if r.GetLoad() > r.maxAllowLoad {
		r.localLock.Lock()
		er := r.lock.Unlock(ctx)
		if er != nil {
			r.l.Warn("分布式锁解锁失败", logger.Error(er))
		}
		r.localLock.Unlock()
	}
	return r.svc.TopN(ctx)
}

func (r *RankingJob) Close() error {
	r.localLock.Lock()
	lock := r.lock
	r.localLock.Unlock()
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	return lock.Unlock(ctx)
}

func (r *RankingJob) RunV1() error {
	// first == 抢分布式锁 全局同一时刻只有一个实例（协程）会运行
	ctx, cancel := context.WithTimeout(context.Background(), time.Second*4)
	defer cancel()
	lock, err := r.client.Lock(ctx, r.key, r.timeout,
		&rlock.FixIntervalRetry{
			Interval: time.Millisecond * 100,
			Max:      3,
			// 重试的超时
		}, time.Second)
	if err != nil {
		return err
	}
	defer func() {
		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
		defer cancel()
		er := lock.Unlock(ctx)
		if er != nil {
			r.l.Error("ranking job释放分布式锁失败", logger.Error(er))
		}
	}()

	//这边就是你拿到了锁 真实业务逻辑
	ctx, cancel = context.WithTimeout(context.Background(), r.timeout)
	defer cancel()

	return r.svc.TopN(ctx)
}
